package com.yanqu.road.reids.subpub;

import com.yanqu.road.logic.redis.RedisManger;
import com.yanqu.road.server.TempMgr;
import com.yanqu.road.utils.date.DateHelper;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class RedisReliableMessageSubPub extends JedisPubSub{

    private static List<ReliableMessageSubPub> list = new ArrayList<>();

    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 2, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<Runnable>(),  new ThreadPoolExecutor.DiscardPolicy());

    private static RedisReliableMessageSubPub factory;

    public static RedisReliableMessageSubPub getInstance(){
        if(factory == null){
            factory = new RedisReliableMessageSubPub();
        }
        return factory;
    }

    public static void addSubPub(ReliableMessageSubPub subPub){
        list.add(subPub);
    }

    @Override
    public void onSubscribe(String channel, int subscribedChannels) {
        TempMgr.getLogger().error("订阅 [{}] 成功", channel);
    }

    @Override
    public void onMessage(String channel, String message) {
        threadPoolExecutor.execute(() -> {
            for (int i = 0; i < list.size(); i++) {
                ReliableMessageSubPub subPub = list.get(i);
                if(subPub.isSupport(channel)){
                    subPub.onMessage(message);
                }
            }
        });
    }

    public long publish(eRedisChannelType channelType, String message){
        ReliableMessageSubPub subPub = getSubPub(channelType);
        if(subPub == null){
            //todo
            return 0;
        }
        String key = subPub.getKey(message);
        boolean exist = RedisManger.getRedisUtil().exists(key);
        if(!exist){
            Jedis jRedis = RedisManger.getRedisUtil().getJRedis();
            jRedis.set(key, message, "NX", "EX", DateHelper.DAY_SECONDS * 3);
            jRedis.close();
            return RedisManger.getRedisUtil().publish(channelType.getChannel(), message);
        }
        return 0;
    }

    public void subscribe(){
        //改为非匿名线程
        RedisSubPubRunner runner = new RedisSubPubRunner(this);
        new Thread(runner).start();
    }

    public static void execute(Runnable runnable){
        threadPoolExecutor.execute(runnable);
    }

    public static ReliableMessageSubPub getSubPub(eRedisChannelType channelType){
        for (ReliableMessageSubPub subPub : list) {
            if(subPub.isSupport(channelType.getChannel())){
                return subPub;
            }
        }
        return null;
    }

    public static List<ReliableMessageSubPub> getList() {
        return list;
    }

    public static int getToday(){
        return Integer.valueOf(DateHelper.getPyyyyMMddDateString(System.currentTimeMillis() / 1000));
    }
}
